package cn.azzhu.day07

import cn.azzhu.day02.{SensorReading, SensorSource}
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala.{StreamTableEnvironment, tableConversions}
import org.apache.flink.types.Row

/**
 * TableFromDataStreamExample
 *
 * @author azzhu
 * @create 2020-09-23 00:16:54
 */
object TableFromDataStreamExample {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val settings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()

    //初始化一个表环境
    val tableEnv = StreamTableEnvironment.create(env, settings)

    val stream:DataStream[SensorReading] = env.addSource(new SensorSource)

    //字段名必须以单引号开始，as用来起别名
    //DataStream => Table
    val table:Table = tableEnv.fromDataStream(stream,'id,'timestamp as 'ts,'temperature)

   // table.select("id").toAppendStream[Row].print()

    //使用数据流创建临时表,目的是做sql查询
    tableEnv.createTemporaryView("sensor",stream)

    tableEnv.sqlQuery("select id from sensor where id = 'sensor_1'")
      .toRetractStream[Row] //撤回流 第一个字段true，表示追加，false表示撤回
      .print()

    //todo DataStream => Table => CRUD => Table =>DataStream => Sink
    env.execute("TableFromDataStreamExample")
  }
}
